package com.leo.chapter12;

import org.junit.Assert;

import java.util.concurrent.BrokenBarrierException;
import java.util.concurrent.CyclicBarrier;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.atomic.AtomicInteger;

/**
 *
 * @author xuexiaolei
 * @version 2017年10月19日
 */
public class PutTakeTest {
    private static final ExecutorService pool = Executors.newCachedThreadPool();
    private final AtomicInteger putSum = new AtomicInteger(0);
    private final AtomicInteger takeSum = new AtomicInteger(0);
    private final CyclicBarrier barrier;
    private final BoundedBuffer<Integer> bb;
    private final int nTrials, nPairs;
    private final BarrierTimer timer;

    public PutTakeTest(int capacity,int nPairs, int nTrials) {
        this.bb = new BoundedBuffer<>(capacity);
        this.nTrials = nTrials;
        this.nPairs = nPairs;
        this.timer = new BarrierTimer();
        this.barrier = new CyclicBarrier(nPairs*2+1, timer);
    }

    public void test(){
        try {
            timer.clear();
            for (int i = 0; i < nPairs; i++) {
                pool.execute(new Producer());
                pool.execute(new Consumer());
            }
            barrier.await();//等待所有线程就绪
            barrier.await();//等待所有线程执行完成
            long nsPerItem = timer.getTime() / (nPairs * (long)nTrials);
            System.out.println("Throughput: " + nsPerItem + " ns/item");
            Assert.assertEquals(putSum.get(), takeSum.get());
        } catch (InterruptedException e) {
            e.printStackTrace();
        } catch (BrokenBarrierException e) {
            e.printStackTrace();
        }
    }

    public static void main(String[] args) {
        new PutTakeTest(10, 10, 100000).test();
        pool.shutdown();
    }

    /**
     * 生产者
     */
    private class Producer implements Runnable {
        @Override public void run() {
            try {
                int seed = (int) (this.hashCode() ^ System.nanoTime());
                int sum = 0;
                barrier.await();
                for (int i = 1; i <= nTrials; i++) {
                    bb.put(seed);
                    sum += seed;
                    seed = xorShift(seed);
                }
                putSum.getAndAdd(sum);
            } catch (InterruptedException e) {
                e.printStackTrace();
            } catch (BrokenBarrierException e) {
                e.printStackTrace();
            }
        }
    }

    /**
     * 消费者
     */
    private class Consumer implements Runnable {
        @Override public void run() {
            try {
                barrier.await();
                int sum = 0;
                for (int i = 1; i < nTrials; i++) {
                    sum += bb.take();
                }
                takeSum.getAndAdd(sum);
                barrier.await();
            } catch (InterruptedException e) {
                e.printStackTrace();
            } catch (BrokenBarrierException e) {
                e.printStackTrace();
            }
        }
    }

    /**
     * 计时器
     */
    class BarrierTimer implements Runnable{
        private boolean started;
        private long startTime,endTime;
        @Override public void run() {
            long t = System.nanoTime();
            if (!started){
                started = true;
                startTime = t;
            }else {
                endTime = t;
            }
        }
        public synchronized void clear(){
            started = false;
        }
        public synchronized long getTime(){
            return endTime - startTime;
        }
    }

    /**
     * 适合测试用的随机数生成器,y可以用nanoTime
     * @param y
     * @return
     */
    static int xorShift(int y){
        y ^= y<<6 ;
        y ^= y>>>21 ;
        y ^= y<<7 ;
        return y;
    }
}
